
#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>

#include <zmq.hpp>

#include "boque/tranzmq/channel.hpp"
#include "boque/tranzmq/client.hpp"

namespace boque		{
namespace tranzmq	{


///////////////////////////////////////////////////////////////////////
//

Client::Client( const std::string& buf_dir )
	:	_buf_dir( buf_dir ),
		_zmq_context_p( NULL )
{
	zmq::context_t*	context_p = new zmq::context_t( 32, 32 );
	_zmq_context_p = context_p;
}

Client::~Client()
{
	if ( NULL != _zmq_context_p ) {
		zmq::context_t*	context_p = static_cast< zmq::context_t * >( _zmq_context_p );
		delete context_p;

		_zmq_context_p = NULL;
	}
}


///////////////////////////////////////////////////////////////////////
//

int
Client::send( const std::string& to, const std::string& data )
{
	channel_v_ptr	channel_vp = getChannel( to );
	Channel*	channel_p = static_cast< Channel* >( channel_vp );

	int	rc = channel_p->send( data );

	return	rc;
}

///////////////////////////////////////////////////////////////////////
//


Client::channel_v_ptr
Client::getChannel( const std::string& addr )
{
	channels_iterator_type	result = _channels.find( addr );
	if ( _channels.end() != result ) {
		// found
		return	result->second;
	}

	std::string	dir_s = addr;
	boost::replace_all( dir_s, ":", "%" );
	boost::replace_all( dir_s, "/", "%" );

	std::string	channel_dir = _buf_dir + "/" + dir_s;

	boost::filesystem::path	p( channel_dir );
	if ( !boost::filesystem::exists( p ) ) {
        try {
            boost::filesystem::create_directories( p );
        }   
        catch ( std::exception& e ) { 
            std::ostringstream  msg;
            msg << "fail to create channel buf directory " << channel_dir << " -- " << e.what();
            throw   std::runtime_error( msg.str() );
        }   
	}

	zmq::context_t*	context_p = static_cast< zmq::context_t* >( _zmq_context_p );
	Channel*	channel_p = new Channel( addr, channel_dir, context_p );

	_channels.insert( std::make_pair( addr, channel_p ) );

	return	channel_p;
}


///////////////////////////////////////////////////////////////////////
//

}	//// namespace tranzmq
}	//// namespace boque

